package br.pucrs.lognplay.job.mapper;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.log4j.Logger;

import br.pucrs.lognplay.job.ElasticSearchClient;

public class IndexerJobMapper extends MapReduceBase implements Mapper<LongWritable, Text, NullWritable, NullWritable> {

	public static final String STATIC_VIEW_ID_PARAMETER = "STATIC_VIEW_ID";

	public static final String OUTPUT_PATH_PARAMETER = "OUTPUT_PATH";

	public static final String ES_DOC_TYPE = "logmsg";

	public static final int ES_BUFFER_SIZE = 100;

	public static final int ES_TTL = 0;

	private String line = null;

	private int linesProcessed = 0;

	private ElasticSearchClient esClient;

	private JobConf job;

	private static final Logger logger = Logger.getLogger(IndexerJobMapper.class);

	public void configure(JobConf job) {
		try {
			this.job = job;

			logger.trace("========= Preparing to run Mapper ==========");
			logger.trace("JobName:      " + job.getJobName());
			logger.trace("staticViewId: " + job.get(STATIC_VIEW_ID_PARAMETER, "lognplay"));
			logger.trace("============================================");

			esClient = new ElasticSearchClient();

			esClient.setIndexName(job.get(STATIC_VIEW_ID_PARAMETER, "lognplay"));

			esClient.startClient();
			esClient.setBufferSize(ES_BUFFER_SIZE);
			esClient.setType(ES_DOC_TYPE);
			esClient.setTtl(ES_TTL);

		} catch (Exception e) {
			logger.error(e);
			e.printStackTrace();
		}
	}

	public void map(LongWritable key, Text value, OutputCollector<NullWritable, NullWritable> output, Reporter reporter) throws IOException {
		line = value.toString();
		try {
			linesProcessed++;
			esClient.put(line);
		} catch (Exception e) {
			logger.error(e);
			e.printStackTrace();
		}
	}

	public void close() throws IOException {
		try {
			logger.trace("============= Report ==============");
			logger.trace("JobName: " + job.getJobName());
			logger.trace("Lines Processed: " + linesProcessed);
			logger.trace("====================================");

			esClient.stopClient();

			// HDFSTools.rm(job.get(OUTPUT_PATH_PARAMETER));
			// HDFSTools.getFileSystem().close();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}
